package com.heyi.nettylong.server.typeServer;

import com.heyi.nettylong.common.protobuf.Command;
import com.heyi.nettylong.common.util.MessageUtil;
import com.heyi.nettylong.server.channelInitializer.ServerChannelInitializer;
import com.heyi.nettylong.server.components.ChannelRepository;
import com.heyi.nettylong.common.MessageRepository;
import com.heyi.nettylong.server.config.ServerProperties;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.concurrent.DefaultEventExecutor;
import io.netty.util.concurrent.DefaultPromise;
import io.netty.util.concurrent.EventExecutor;
import io.netty.util.concurrent.Promise;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;

/**
 * Netty 服务类
 * @author dengfeng
 */
public class NettyServer extends Thread {
	private final Logger log = LoggerFactory.getLogger(this.getClass());

	public static ChannelRepository channelRepository = new ChannelRepository();

	ServerProperties serverProperties;

	private final boolean keepAlive=true;

	/**
	 * BACKLOG用于构造服务端套接字ServerSocket对象，标识当服务器请求处理线程全满时，
	 * 用于临时存放已完成三次握手的请求的队列的最大长度。如果未设置或所设置的值小于1，Java将使用默认值50
	 */
	private final int backlog=128;

	public NettyServer(ServerProperties serverProperties){
		this.serverProperties = serverProperties;
	}

	/**用来监控tcp链接 指定线程数 默认是1 用默认即可*/
	private NioEventLoopGroup bossGroup() {
		return new NioEventLoopGroup(serverProperties.getBossCount());
	}

	/**处理io事件 一定要多线程效率高 源码中默认是cpu核数*2 */
	private NioEventLoopGroup workerGroup() {
		if(serverProperties.getWorkerCount() == 0){
			return new NioEventLoopGroup();
		}else {
			return new NioEventLoopGroup(serverProperties.getWorkerCount());
		}
	}

	/**
	 * netty服务器相关设置
	 */
	private Map<ChannelOption<?>, Object> tcpChannelOptions() {
		Map<ChannelOption<?>, Object> options = new HashMap<ChannelOption<?>, Object>(100);

        /*是否启用心跳保活机制。在双方TCP套接字建立连接后（即都进入ESTABLISHED状态）
        并且在两个小时左右上层没有任何数据传输的情况下，这套机制才会被激活
        */
		options.put(ChannelOption.SO_KEEPALIVE, keepAlive);

        /*
        在TCP/IP协议中，无论发送多少数据，总是要在数据前面加上协议头，
        同时，对方接收到数据，也需要发送ACK表示确认。
        为了尽可能的利用网络带宽，TCP总是希望尽可能的发送足够大的数据。
        这里就涉及到一个名为Nagle的算法，该算法的目的就是为了尽可能发送大块数据，避免网络中充斥着许多小数据块。
        */
		options.put(ChannelOption.TCP_NODELAY,true);

		return options;
	}

	/**
	 * netty服务器启动帮助类
	 */
	private ServerBootstrap getServerBootstrap() {
		ServerBootstrap b = new ServerBootstrap();
		b.group(bossGroup(), workerGroup())
				.channel(NioServerSocketChannel.class)
				.handler(new LoggingHandler(LogLevel.DEBUG))
				.option(ChannelOption.SO_BACKLOG,backlog)
				.childHandler(new ServerChannelInitializer(serverProperties));
		Map<ChannelOption<?>, Object> tcpChannelOptions = tcpChannelOptions();
		Set<ChannelOption<?>> keySet = tcpChannelOptions.keySet();
		for (ChannelOption option : keySet) {
			b.childOption(option, tcpChannelOptions.get(option));
		}
		return b;
	}

	public InetSocketAddress getTcpPort() {
		return new InetSocketAddress(serverProperties.getHost(),serverProperties.getPort());
	}

	@Override
	public void run() {
		try {
			getServerBootstrap().bind(getTcpPort()).sync().channel().closeFuture().sync();
		} catch (InterruptedException e) {
			log.error("启动Netty服务出错:"+e.getMessage());
			e.printStackTrace();
		}
	}

	/**
	 * 发送消息
	 * @param clientId 客户端IFD
	 * @param text 消息内容
	 * @return
	 * @throws Exception
	 */
	public ChannelFuture sendMessage(String clientId, String text) throws Exception {
		Channel channel = channelRepository.get(clientId);
		// 判断是否活跃
		if(channel==null){
			throw new Exception("客户端["+clientId+"]未连接或已断线");
		}
		if(!channel.isActive()){
			channelRepository.remove(clientId);
			throw new Exception("客户端["+clientId+"]连接状态异常：isActive=false");
		}
		return channel.writeAndFlush(MessageUtil.build(clientId, Command.CommandType.PUSH_DATA, text));
	}

	/**
	 * 发送消息并得到返回的消息(阻塞方法，谨慎使用)
	 * @param clientId 客户端IFD
	 * @param text 消息内容
	 * @param timeout 等待客户端返回消息的时间，单位：秒
	 * @return 客户返回的消息
	 * @throws Exception
	 */
	public synchronized String sendMessageAndReturn(String clientId, String text, long timeout) throws Exception {
		Channel channel = channelRepository.get(clientId);
		// 判断是否活跃
		if (channel == null) {
			throw new Exception("客户端[" + clientId + "]未连接或已断线");
		}
		if (!channel.isActive()) {
			channelRepository.remove(clientId);
			throw new Exception("客户端[" + clientId + "]连接状态isActive异常");
		}
		int msgId = MessageRepository.buildSequence();
		// 构造线程池
		EventExecutor executor = new DefaultEventExecutor();
		// 创建 DefaultPromise 实例
		Promise promise = new DefaultPromise(executor);
		MessageRepository.addListener(msgId, (String content) ->
				promise.setSuccess(content));

		channel.writeAndFlush(MessageUtil.build(msgId, clientId, Command.CommandType.PUSH_DATA, text));

		String msg = null;
		try{
			msg = promise.get(timeout, TimeUnit.SECONDS).toString();
		}catch (Exception ex){
			MessageRepository.removeListener(msgId);
			throw new Exception("客户端返回消息超时");
		}
		return msg;
	}
}
